package cc.autoapi.pucong.auto.flow.core.flow.execNode.streamBus;


import cc.autoapi.pucong.auto.flow.core.flow.execNode.utils.FlowThreadPoolUtils;

import java.util.ArrayList;
import java.util.List;

/**
 * 订阅者
 */
public class MsgStreamSubscriber implements IStreamSubscriber {

    private IStreamServiceEmitter IStreamServiceEmitter;
    private IStreamServiceComplate iStreamServiceComplate;

    private List<String> msgList = new ArrayList<>();

    @Override
    public void update(String message) {
        if (IStreamServiceEmitter != null) {
            if (msgList.size() > 0) {
                for (String s : msgList) {
                    IStreamServiceEmitter.run(s);
                    msgList.remove(s);
                }
            }
            IStreamServiceEmitter.run(message);
        } else {
            msgList.add(message);
        }
    }

    @Override
    public void emitter(IStreamServiceEmitter IStreamServiceEmitter, IStreamServiceComplate iStreamServiceComplate) {
        this.IStreamServiceEmitter = IStreamServiceEmitter;
        this.iStreamServiceComplate = iStreamServiceComplate;
    }

    @Override
    public void runComplete() {
        if (IStreamServiceEmitter != null) {
            if (msgList.size() > 0) {
                for (String s : msgList) {
                    msgList.remove(s);
                    IStreamServiceEmitter.run(s);
                }
            }
            iStreamServiceComplate.complete();
        } else {
            // 等待300ms后再次执行
            FlowThreadPoolUtils.executor(() -> {
                try {
                    Thread.sleep(300);
                } catch (InterruptedException e) {
                }
                this.runComplete();
            });
        }
    }
}